package edu.pitt.myptm.scheduler;
import edu.pitt.myptm.AbortCommand;
import edu.pitt.myptm.CommandFactory;
import edu.pitt.myptm.MyCommand;

import java.io.IOException;
import java.util.*;
import java.util.logging.FileHandler;
import java.util.logging.Logger;
import java.util.logging.SimpleFormatter;

import org.jgrapht.DirectedGraph;
import org.jgrapht.Graph;
import org.jgrapht.graph.DefaultDirectedGraph;
import org.jgrapht.graph.DefaultEdge;

import edu.pitt.myptm.datamanager.DataManager;
import edu.pitt.myptm.transactionmanger.TransactionManager;

public class Scheduler {
		private Logger log = Logger.getLogger(Scheduler.class.getName());
	
		static int number_of_commit=0;
		static int number_of_abort=0;
		static int number_of_read=0;
		static int number_of_write=0;
		static int number_of_operation=0;
		static float total_time=0;
		
//		TransactionManager tm = new TransactionManager();
		
		//pull a command from transaction manager
		public MyCommand pull_operation(TransactionManager tm){
//			System.out.println("Pull 1 operation from Transaction Manager");
			MyCommand command=null;
			try {
//				System.out.println("hello1");
				command=tm.loadNext();//.loadRNext();
//				System.out.println("hello2");
				if(command!=null){
					System.out.println("Pull: TranID:"+command.getTransactionId()+
											" Opt_Type:"+command.getType()+
											" File:"+command.getFilename()+
											" timestamp:"+ command.getTimeStamp()+ "\n");
					if(command.isReadCommand())
						number_of_read++;
					else if(command.isWriteCommand())
						number_of_write++;
					
					number_of_operation++;
				}
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			return command;
		}
		
		//push a command to data manager
		public void push_operation(MyCommand cmd,DataManager DM){
			log.info("Push 1 operation to Data Manager");
			//push the command to Data Manager to execute
			log.info("TranID="+cmd.getTransactionId()+" type:"+cmd.getType()
								+" "+cmd.getFilename()+" row="+cmd.getId()+" timestamp:"+ cmd.getTimeStamp()+ "\n");
			
			
			if(cmd.isCommitCommand())
				number_of_commit++;
			if(cmd.isAbortCommand())
				number_of_abort++;
			DM.execute(cmd);
		}
		
		public ArrayList<MyCommand> sort_execute_list(ArrayList<MyCommand> execute_list){
			
//			Arrays.sort(execute_list);
			Collections.sort(execute_list);
			return execute_list;
		}
		
		public MyCommand remove_current_commit_from_commit_list(ArrayList<MyCommand> commit_list,MyCommand cmd){
			MyCommand c=new MyCommand();
			for(int i=0; i<commit_list.size(); i++){
				if(commit_list.get(i).getId()==cmd.getTransactionId()&&
						commit_list.get(i).getFilename().equals(cmd.getFilename())){
					c=commit_list.get(i);
					commit_list.remove(i);
					break;
				}
			}
			return c;
		}
		
		public void CommandHandler(MyCommand cmd,Hashtable<LockTuple,LockContent> lock_table,
										Hashtable<Integer, LockGranted> lock_holder_waitting_list,
										CompatibleTable ComTable,ArrayList<MyCommand> commit_list,
										DefaultDirectedGraph<Integer, DefaultEdge> WFG, LockManager LM, DeadLockManager DLM,
										Hashtable<Integer,Integer> abortList,Hashtable<String,LockContent> file_lock_table,
										DataManager DM, Hashtable<Integer,Integer> TranTable,
										Hashtable<Integer,Integer> ProcessTable){
			boolean LockObtained=false;
			if(cmd.isBeginCommand()){
				System.out.println("isTransaction="+cmd.isTransaction()+" "+cmd.getTransactionId());
				if(cmd.isTransaction()){
					TranTable.put(cmd.getTransactionId(),1);
				}else{
					ProcessTable.put(cmd.getTransactionId(),1);
				}
				System.out.println("TranTable size="+TranTable.size()+ " ProcessTable size="+ProcessTable.size());
			} else if(cmd.isReadCommand()||cmd.isWriteCommand()||cmd.isDeleteCommand()||cmd.isMatchCommand()){
				//WFG updated in AcquireLock
				LockObtained=LM.AcquireLock(cmd,lock_table,lock_holder_waitting_list,ComTable,false,WFG,DLM,file_lock_table,
											TranTable,ProcessTable);
				
//				DLM.updateWFG(WFG);
				//2 reasons for LockObtaied=false
				//1. deadlock; 2. waiting for the lock
				if(LockObtained==true){//if can obtain lock, should be no deadlock
					push_operation(cmd,DM);
				} else {
					if(DLM.DetectDeadlock(WFG)==true){
//					if(false){
						int victimTran=DLM.SelectVictim(WFG);
						System.out.println("DeadLock! Abort Transaction "+victimTran);
						log.info("DeadLock! Abort Transaction "+victimTran);
						abortList.put(victimTran, 1);
						
						//abort the victimTran
						//do I have to check the lock_table waiting list after resolve deadlock??
						AbortCommand cmd1 = (AbortCommand)CommandFactory.makeCommand(MyCommand.ABORT);
						cmd1.setTransactionId(victimTran);
						LM.WFGDeleteEdge(WFG,cmd);//delete all outgoing edge of the victim
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
						//just copy the code from abort
//						CommandHandler(cmd1,lock_table,lock_holder_waitting_list,commit_list,WFG,LM,DLM);
						//LM.ReleaseLock also checked what operation(s) can acquire lock after someone releases lock(s)
						ArrayList<MyCommand> execute_list= LM.ReleaseLock(cmd1,lock_table,lock_holder_waitting_list,ComTable,
																	WFG,DLM,true,file_lock_table,TranTable,ProcessTable);
						push_operation(cmd1,DM);
						lock_holder_waitting_list.remove(victimTran);
						
						if(execute_list!=null){
							execute_list=sort_execute_list(execute_list);
							for(int i=0; i< execute_list.size();i++)
								push_operation(execute_list.get(i),DM);
						}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
					}
				}
			} else if(cmd.isCommitCommand()) {//*only time that waiting tran can obtain lock is when someone commit
				if(LM.check_all_prior_operation_finish(cmd,lock_table,lock_holder_waitting_list,ComTable,file_lock_table)){
//				if(LM.check_all_prior_operation_finish(cmd,lock_table,lock_holder_waitting_list,ComTable,file_lock_table)){
					//LM.ReleaseLock also checked what operation(s) can acquire lock after someone releases lock(s)
					ArrayList<MyCommand> execute_list= LM.ReleaseLock(cmd,lock_table,lock_holder_waitting_list,ComTable,WFG,
																	DLM,false,file_lock_table,TranTable,ProcessTable);
					push_operation(cmd,DM);
					
					if(execute_list!=null){
						execute_list=sort_execute_list(execute_list);
						for(int i=0; i< execute_list.size();i++)
							push_operation(execute_list.get(i),DM);
					}
					
					//check_commit_list see any transaction can commit
					int size=commit_list.size();
					System.out.println("commit_list.size()="+commit_list.size());
//					for(int i=0; i<size; i++){
					int i=0,s=0;
					while(commit_list.size()!=0&&i<commit_list.size()&&i>=0){
						System.out.println("i="+i);
						if(LM.check_all_prior_operation_finish(commit_list.get(i),lock_table,lock_holder_waitting_list,ComTable,file_lock_table)){
							//LM.ReleaseLock also checked what operation(s) can acquire lock after someone commit and releases lock(s)
							ArrayList<MyCommand> execute_list2= LM.ReleaseLock(commit_list.get(i),lock_table,lock_holder_waitting_list,ComTable,WFG,
																				DLM,false,file_lock_table,TranTable,ProcessTable);
							System.out.println("\n com list+++++TranID="+commit_list.get(i).getTransactionId()
									  +" type:"+commit_list.get(i).getType()
									  +" "+commit_list.get(i).getFilename()
									  +" row="+commit_list.get(i).getId());
							System.out.println("execute_list2="+execute_list2.size());
							push_operation(commit_list.get(i),DM);
							commit_list.remove(i);
							i--; //size--;
							
							if(execute_list2!=null){
								execute_list2=sort_execute_list(execute_list2);
								for(int j=0; j< execute_list2.size();j++)
									push_operation(execute_list2.get(j),DM);
							}
						}else
							//s++;
							i++;
					}
					//may need changes here!!!!
					System.out.println("realease lock: commit list end commit_list.size()="+commit_list.size());
					int size2=commit_list.size();
					for(int i1=0; i1<size2; i1++){
						if(LM.check_all_prior_operation_finish(commit_list.get(i1),lock_table,lock_holder_waitting_list,ComTable,file_lock_table)){
							ArrayList<MyCommand> execute_list2= LM.ReleaseLock(commit_list.get(i1),lock_table,lock_holder_waitting_list,ComTable,WFG,
																			DLM,false,file_lock_table,TranTable,ProcessTable);
							push_operation(commit_list.get(i1),DM);
							commit_list.remove(i1);
							i1--; size2--;
						}
					}
					System.out.println("realease lock: commit list end commit_list.size()="+commit_list.size());
				} else {
					//put the commit command into commit_list
					System.out.println("Put into commit_list TranID="+cmd.getTransactionId()
															  +" type:"+cmd.getType()
															  +" "+cmd.getFilename()
															  +" row="+cmd.getId());
					commit_list.add(cmd);
					System.out.println("commit_list.size()="+commit_list.size());
				}
			} else if(cmd.isAbortCommand()){
				//LM.ReleaseLock also checked what operation(s) can acquire lock after someone releases lock(s)
				ArrayList<MyCommand> execute_list= LM.ReleaseLock(cmd,lock_table,lock_holder_waitting_list,ComTable,WFG,DLM,true,
																file_lock_table,TranTable,ProcessTable);
				push_operation(cmd,DM);
				
				if(execute_list!=null){
					execute_list=sort_execute_list(execute_list);
					for(int i=0; i< execute_list.size();i++)
						push_operation(execute_list.get(i),DM);
				}
			} /*else if(cmd.isMatchCommand()){
				//WFG updated in AcquireLock
				LockObtained=LM.AcquireLock(cmd,lock_table,lock_holder_waitting_list,ComTable,false,WFG,DLM,file_lock_table);
			} else if(cmd.isDeleteCommand()){
				
			}*/
		}
		
		public boolean CheckAbortedList(MyCommand cmd,Hashtable<Integer,Integer> abortList){
			if(abortList.get(cmd.getTransactionId())!=null){
				System.out.println("Aborted!");
				return false;
			}else
				return true;
		}
		
		public void init_compatibleTable(CompatibleTable comTable){
			//0:read 1:write
			comTable.CompatibleTable[0][0]=1;
			comTable.CompatibleTable[0][1]=0;
			comTable.CompatibleTable[0][2]=1;
			comTable.CompatibleTable[0][3]=0;
			comTable.CompatibleTable[0][4]=1;
			comTable.CompatibleTable[0][5]=0;
			comTable.CompatibleTable[0][6]=1;
			comTable.CompatibleTable[0][7]=0;
			
			comTable.CompatibleTable[1][0]=0;
			comTable.CompatibleTable[1][1]=0;
			comTable.CompatibleTable[1][2]=0;
			comTable.CompatibleTable[1][3]=0;
			comTable.CompatibleTable[1][4]=1;
			comTable.CompatibleTable[1][5]=0;
			comTable.CompatibleTable[1][6]=1;
			comTable.CompatibleTable[1][7]=0;
			
			comTable.CompatibleTable[2][0]=1;
			comTable.CompatibleTable[2][1]=0;
			comTable.CompatibleTable[2][2]=1;
			comTable.CompatibleTable[2][3]=0;
			comTable.CompatibleTable[2][4]=1;
			comTable.CompatibleTable[2][5]=0;
			comTable.CompatibleTable[2][6]=1;
			comTable.CompatibleTable[2][7]=0;
			
			comTable.CompatibleTable[3][0]=0;
			comTable.CompatibleTable[3][1]=0;
			comTable.CompatibleTable[3][2]=0;
			comTable.CompatibleTable[3][3]=0;
			comTable.CompatibleTable[3][4]=1;
			comTable.CompatibleTable[3][5]=0;
			comTable.CompatibleTable[3][6]=1;
			comTable.CompatibleTable[3][7]=0;
			
			comTable.CompatibleTable[4][0]=1;
			comTable.CompatibleTable[4][1]=1;
			comTable.CompatibleTable[4][2]=1;
			comTable.CompatibleTable[4][3]=1;
			comTable.CompatibleTable[4][4]=1;
			comTable.CompatibleTable[4][5]=1;
			comTable.CompatibleTable[4][6]=1;
			comTable.CompatibleTable[4][7]=1;
			
			comTable.CompatibleTable[5][0]=0;
			comTable.CompatibleTable[5][1]=0;
			comTable.CompatibleTable[5][2]=0;
			comTable.CompatibleTable[5][3]=0;
			comTable.CompatibleTable[5][4]=1;
			comTable.CompatibleTable[5][5]=1;
			comTable.CompatibleTable[5][6]=1;
			comTable.CompatibleTable[5][7]=1;
			
			comTable.CompatibleTable[6][0]=1;
			comTable.CompatibleTable[6][1]=1;
			comTable.CompatibleTable[6][2]=1;
			comTable.CompatibleTable[6][3]=1;
			comTable.CompatibleTable[6][4]=1;
			comTable.CompatibleTable[6][5]=1;
			comTable.CompatibleTable[6][6]=1;
			comTable.CompatibleTable[6][7]=1;
			
			comTable.CompatibleTable[7][0]=0;
			comTable.CompatibleTable[7][1]=0;
			comTable.CompatibleTable[7][2]=0;
			comTable.CompatibleTable[7][3]=0;
			comTable.CompatibleTable[7][4]=1;
			comTable.CompatibleTable[7][5]=1;
			comTable.CompatibleTable[7][6]=1;
			comTable.CompatibleTable[7][7]=1;
		}
		
		public void Schedule_Transactions(TransactionManager tm,DataManager DM){
			
			try {
				FileHandler filehandler  = new FileHandler("scheduler.log",true);
				filehandler.setFormatter(new SimpleFormatter());
				log.addHandler(filehandler);
			} catch (SecurityException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			System.out.println("STart scheduling transaction");
			CompatibleTable comTable=new CompatibleTable();
//			Graph WFG=new Graph();
//			Vertex V=new Vertex();
			DefaultDirectedGraph<Integer, DefaultEdge> WFG = new DefaultDirectedGraph<Integer, DefaultEdge>(DefaultEdge.class);
			LockManager LM=new LockManager();
			DeadLockManager DLM=new DeadLockManager();
			Hashtable<Integer,Integer> abortList= new Hashtable<Integer,Integer>();
			Hashtable<LockTuple,LockContent> lock_table= new Hashtable<LockTuple,LockContent>();
			Hashtable<String,LockContent> file_lock_table= new Hashtable<String,LockContent>();
			Hashtable<Integer, LockGranted> lock_holder_waitting_list= new Hashtable<Integer, LockGranted>();
			ArrayList<MyCommand> commit_list=new ArrayList<MyCommand>();			
//			DataManager DM=new DataManager();
			Hashtable<Integer,Integer> TranTable=new Hashtable<Integer,Integer>();
			Hashtable<Integer,Integer> ProcessTable=new Hashtable<Integer,Integer>();
			
			init_compatibleTable(comTable);
			
			float startTime=System.nanoTime();//.currentTimeMillis();
			MyCommand command=pull_operation(tm);
			while(command!=null) {
				if(CheckAbortedList(command,abortList)==true){//no need to handle all aborted transaction operation
					//commandHandler return which command to execute next
					CommandHandler(command,lock_table,lock_holder_waitting_list,comTable,commit_list,WFG,LM,DLM,abortList
									,file_lock_table,DM,TranTable,ProcessTable);
				}
				System.out.println("");
				command=pull_operation(tm);
			}
			float endTime=System.nanoTime();//.currentTimeMillis();
			total_time=endTime-startTime;
			log.info("number of commit="+ number_of_commit+ " number_of_abort="+number_of_abort + " number of read="+number_of_read
					+" nnumber_of_write="+number_of_write+" total_number_of_operation="+number_of_operation
					+ " average response time="+total_time/number_of_operation +" nano second");
		}
}
